package com.k2data.processors;
import com.k2data.common.Constants;
import com.k2data.service.SchemaService;
import org.apache.avro.Schema;
import org.apache.kafka.common.network.SslChannelBuilder;
import org.apache.kafka.common.requests.ProduceResponse;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
import org.json.JSONArray;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * Created by luoqifei on 17-1-12.
 * for lambda,need jdk 8
 */
public class TransformProcessor implements Processor<String, String> {
    private static Logger logger = LoggerFactory.getLogger(TransformProcessor.class);
    private ProcessorContext context;
    private SchemaService schemaService;
    private  Map<String,String> sensor;
    private String fieldGroupId = "hn_001";
    private String[] strRecords = null;
    private String startTag = "realtimedata|";
    private String wfid = "";
    private String wtid = "";
    private int wfidLength = 6;
    private int wtidLength =3 ;
    private List<String> fieldNameList = new ArrayList<>();
    //private KeyValueStore kvStore;
    public  String testRecord = startTag+"|510401059,2017-01-12 20:19:51.227,236.08,5.22,260.60,1.37,-0.11,36.55,36.55,36.55,3447.00,151.89,136.76,12.92,12.88,12.92,12.92,0.04,0.01,0.01,0.02,38.79,38.59,37.59,37.79,39.29,37.90,39.20,344.00,346.69,347.17,201.71,217.53,195.12,215.08,49.96,3.41,0.99,23.39,22.60,16.60,0.00,16.60,18.00,18.20,17.89,23.39,23.79,23.60,20.79,21.29,21.29,19.60,20.10,20.10,20.79,21.29,21.29,58.71,59.04,58.76,29.32,29.44,29.32,0.10,0.10,0.10,0.00,-0.07,-0.03,0.00,0.00,0.00,8937.10,8685.69,192.51,7329.01,7036.55,2495714.000,2106.00,57.90,0.00,21.85,21.95,21.61,21.36,25.02,25.22,25.32,25.22,25.37,25.07,25.37,21.95,24.20,21.79,51.79,50.29,38.90,24.60,20.50,551.67,-551.33,146.12,3.25,874.68,0.00,0.00,60.00,0.00,23.39,600,0,0,0,6640.52,1617.35,0.10,1.00,1608.10,0.00,0.00,6368.73,0,21.79,22.29,850.00,0.00,0.00,0,1500,298,0,0,22123,22123,22123,52,2017,1,12,20,14,19,5,0,0,0,0,0,0,0,0,0,0,0,0,0,170111,2326,161127,912,0,0,0,0,0,0,0,0,0,0,0,3,1,3,TRUE,FALSE,FALSE,FALSE,FALSE,TRUE,FALSE,TRUE,FALSE,FALSE,FALSE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,TRUE,FALSE,FALSE,FALSE,TRUE,TRUE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,TRUE,TRUE,FALSE,TRUE,TRUE,TRUE,FALSE,TRUE,TRUE,TRUE,FALSE,FALSE,TRUE,TRUE,TRUE,TRUE,TRUE,TRUE,FALSE,FALSE,FALSE,0,0,2017-01-12 20:14:19,2017-01-11 23:26:00,2016-11-27 09:12:00,True,False,False,False,False,False,FALSE,FALSE,FALSE,FALSE,TRUE,TRUE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,111160128,0,0,0,0,0,0,0,0,80,2495714.000,2106,0,0,0,0,0,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,FALSE,0.00,0.00,0,0,0,99.40,5.20,5.86,215,0,1,FALSE";
    @Override
    @SuppressWarnings("unchecked")
    public void init(ProcessorContext context) {
        this.context = context;
        //interval 60 seconds push the kv store to next processor
        this.context.schedule(60000);
        // start schemaService
        try {
            if (schemaService == null) {
                schemaService = new SchemaService(KmxTopology.zkUrl,KmxTopology.zkNs);
            }
            schemaService.start();
            // sync with zk when preparing bolt task
            schemaService.reload();
        } catch (Exception e) {
            logger.error("Fail to start schemaService: {}", e.getMessage());
            e.printStackTrace();
            System.exit(1);
        }
        //init kv state store
        //this.kvStore = (KeyValueStore)context.getStateStore(Constants.ANOMALY_STATE_STORE_NAME);

        String schemaId = schemaService.getSchemaId(fieldGroupId);
        //get id fields
        //String idFields = schemaService.getIdFieldsByFieldGroupId(fieldGroupId);
        //get sensor filed name and type
        sensor = schemaService.getAllSensorDataTypes(schemaId);
        Schema schema =schemaService.getSchema(schemaId);
        //sub list,remove the field group id,and timestamp
        List<Schema.Field> fields = schema.getFields().subList(2,schema.getFields().size());
        for(Schema.Field field :fields){
            fieldNameList.add(field.aliases().toString().replace("[","").replace("]",""));
        }
    }

    @Override
    public void process(String key, String value) {
        logger.info("processed record key is {}, value is : {}: ",key,value);
        //send to next sink or processor
        try {
            this.context.forward(key,csvRecord2Json(setIdfields(value)).toString());
        }catch (Exception e){
            logger.error("record can't trans to json object. key is {}, value is : {}: ",key,value);
            e.printStackTrace();
            //this.kvStore.put(key,value);
        }
    }

    @Override
    public void punctuate(long timestamp) {
        //push the kv store to next processor
//        KeyValueIterator<String,String> iter = this.kvStore.all();
//        while (iter.hasNext()){
//            KeyValue<String,String> entry = iter.next();
//            context.forward(entry.key,entry.value.toString());
//        }
//        iter.close();
//        context.commit();
    }

    @Override
    public void close() {
        schemaService.close();
        //this.kvStore.close();
    }

    /**
     * get the wfid,wtid,and set the type default is huineng customer send time,such as 2017-01-12 20:19:51.227
     * @param scvRecord
     * @return
     */
    private String setIdfields(String scvRecord){
        //remove the start tag
        String str = scvRecord.substring(scvRecord.indexOf(startTag)+startTag.length()+1,scvRecord.length());
       // str = str.replace("|",",");
        //String idFields = str.substring(0,str.indexOf(",",str.indexOf(",")+1));
        //String strIdFields = idFields.split(",");
        //String timestamp = strIdFields[1];
        String idFields = str.substring(0,str.indexOf(","));
        //remove wfid+wtid,but state the sedment time as 3th column---type
        String str2= str.replaceFirst(idFields+",","");
        wfid= idFields.substring(0,wfidLength);
        wtid =idFields.substring(0,idFields.indexOf("|"));
        return wfid+","+wtid+","+str2;
    }
    /**
     * the strRecord must contain every id field witch isIdField is true.
     * @param strRecord
     */
    public JSONObject csvRecord2Json(String strRecord){
        strRecords = strRecord.split(",");
        JSONObject kmxJsonRecord = new JSONObject();
        //add field group id
        kmxJsonRecord.put(Constants.KMX_JSON_FIELD_GROUP_ID_NAME,fieldGroupId);
        //add timestamp
        JSONObject timestamp = new JSONObject();
        //this timestamp is
        timestamp.put("timestamp",System.currentTimeMillis());
        kmxJsonRecord.put(Constants.KMX_JSON_SAMPLETIME_NAME,timestamp);

        //add fields
        JSONArray fieldJsonArray = new JSONArray();
        for(int i=0; i<strRecords.length; i++){
            JSONObject jsonObject = new JSONObject();
            if(i>=3){
                jsonObject.put(Constants.KMX_JSON_FIELD_ID_NAME, fieldNameList.get(i));
                jsonObject.put("isIdField",false);
                switch (sensor.get(fieldNameList.get(i))) {
                    case "string":
                        jsonObject.put(Constants.KMX_JSON_FIELD_VALUE_NAME,strRecords[i]);
                        break;
                    case "boolean":
                        jsonObject.put(Constants.KMX_JSON_FIELD_VALUE_NAME,Boolean.valueOf(strRecords[i].toLowerCase()));
                        break;
                    case "double":
                        jsonObject.put(Constants.KMX_JSON_FIELD_VALUE_NAME,Double.valueOf(strRecords[i]));
                        break;
                }
            }else {
                switch (i){
                    //first id field is wfid
                    case 0:
                        jsonObject.put(Constants.KMX_JSON_FIELD_ID_NAME, "wfid");
                        jsonObject.put(Constants.KMX_JSON_FIELD_VALUE_NAME,strRecords[i]);
                        jsonObject.put("isIdField",true);
                        break;
                    //2th id field is wtid
                    case 1:
                        jsonObject.put(Constants.KMX_JSON_FIELD_ID_NAME, "wtid");
                        jsonObject.put(Constants.KMX_JSON_FIELD_VALUE_NAME,strRecords[i]);
                        jsonObject.put("isIdField",true);
                        break;
                    //3th id field is type
                    case 2:
                        jsonObject.put(Constants.KMX_JSON_FIELD_ID_NAME, "type");
                        jsonObject.put(Constants.KMX_JSON_FIELD_VALUE_NAME,Constants.HUINENG_ID_FILED_TYPE_VALUE);
                        jsonObject.put("isIdField",true);
                        break;
                }
            }
            fieldJsonArray.put(jsonObject);
        }
        kmxJsonRecord.put(Constants.KMX_JSON_FIELDS_NAME,fieldJsonArray);
        logger.info("trans to json success.\n "+kmxJsonRecord.toString());
        return kmxJsonRecord;
    }
}

